package org.qiyu.live.id.generate.provider.service.impl;

import jakarta.annotation.Resource;
import org.qiyu.live.id.generate.provider.dao.mapper.IdGenerateMapper;
import org.qiyu.live.id.generate.provider.dao.po.IdGeneratePO;
import org.qiyu.live.id.generate.provider.service.IdGenerateService;
import org.qiyu.live.id.generate.provider.service.bo.LocalSeqIdBO;
import org.qiyu.live.id.generate.provider.service.bo.LocalUnSeqIdBO;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.stereotype.Service;

import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicLong;

/**
 * Created with IntelliJ IDEA.
 * Description:
 * User: sfx
 * Date: 2023-09-19
 * Time: 21:35
 */
@Service
public class IdGenerateServiceImpl implements IdGenerateService, InitializingBean {

    private static Logger LOGGER = LoggerFactory.getLogger(IdGenerateServiceImpl.class);

    /**
     * 管理分布式id生成策略
     * 分布式id生成策略,需要再本地内存中的一个map对象进行管理,
     * 1. 存放有序id
     */
    private static Map<Integer, LocalSeqIdBO> localSeqIdBOMap = new ConcurrentHashMap<>();
    /**
     * 2. 存放无序id
     */
    private static Map<Integer, LocalUnSeqIdBO> localUnSeqIdBOMap = new ConcurrentHashMap<>();

    @Resource
    private IdGenerateMapper idGenerateMapper;

    /**
     * 创建一个异步的线程池去执行
     */
    ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(8, 16, 3, TimeUnit.SECONDS, new ArrayBlockingQueue<>(1000), new ThreadFactory() {
        @Override
        public Thread newThread(Runnable run) {
            Thread thread = new Thread(run);
            thread.setName("id-generate-thread" + ThreadLocalRandom.current().nextInt(1000));
            return thread;
        }
    });

    /**
     * 进行限流
     */
    private static Map<Integer, Semaphore> semaphoreMap = new ConcurrentHashMap<>();

    /**
     * 就会尝试更新本地的id段 (有序id生成)
     */
    private static final double UPDATE_SEQ_ID_RATE = 0.75;

    /**
     * 就会尝试更新本地的id段 (无序id生成)
     */
    private static final double UPDATE_UN_SEQ_ID_RATE = 0.25;
    /**
     * 有序id生成
     */
    private static final int ORDER_ID_TYPE = 1;

    /**
     * 生成有序id
     *
     * @param id 对应着一种分布式id生成的策略
     * @return 返回有序的唯一的递增的分布式id
     */
    @Override
    public Long getSeqId(Integer id) {
        // 1. 校验参数
        if (id == null) {
            LOGGER.error("[getSeqId] id is error id is {}", id);
            return null;
        }
        // 2. 根据id生成策略,来得到当前的分布式id
        // 我们需要根据id生成策略获取到对应的PO对象
        LocalSeqIdBO localSeqIdBO = localSeqIdBOMap.get(id);
        if (localSeqIdBO == null) {
            LOGGER.error("[getSeqId]  localSeqIdBO is error id is {}", id);
            return null;
        }
        // 当id段用完了之后,我们尝试进行刷新id段,又因为是在高并发的场景下,我们不能再id段使用完了再去尝试更新
        // 因为尝试更新操作会涉及到很多的update操作,select 会耗费很多的网络IO,所以我们固定,当 当前id段使用了
        // 75%的时候,我们就尝试进行更新
        this.refreshLocalSeqId(localSeqIdBO);
        long returnId = localSeqIdBO.getCurrentNum().incrementAndGet();
        if(returnId > localSeqIdBO.getNextThreshold()) {
            LOGGER.error("[getUnSeqId] id is over limit,id is {}", id);
            return null;
        }
        return returnId;
    }

    /**
     * 生成无序id
     *
     * @param id 分布式id生成的策略
     * @return 返回有序的唯一的递增的分布式id
     */
    @Override
    public Long getUnSeqId(Integer id) {
        // 1. 校验参数
        if (id == null) {
            LOGGER.error("[getUnSeqId] id is error id is {}", id);
            return null;
        }
        // 2. 根据id生成策略,来得到当前的分布式id
        // 我们需要根据id生成策略获取到对应的PO对象
        LocalUnSeqIdBO localUnSeqIdBO = localUnSeqIdBOMap.get(id);
        if (localUnSeqIdBO == null) {
            LOGGER.error("[getUnSeqId]  localSeqIdBO is error id is {}", id);
            return null;
        }
        // 如果本地id段用完了就需要尝试进行刷新本地id段
        this.refreshLocalUnSeqId(localUnSeqIdBO);
        // 取出id返回
        Long returnId = localUnSeqIdBO.getIdQueue().poll();
        if (returnId == null) {
            LOGGER.error("[getUnSeqId] returnId is null,id is {}", id);
            return null;
        }
        return returnId;
    }

    private void refreshLocalUnSeqId(LocalUnSeqIdBO localUnSeqIdBO) {
        // 得到当前id段总占用情况
        long step = localUnSeqIdBO.getNextThreshold() - localUnSeqIdBO.getCurrentStart();
        // 还有一种情况,假设并发量是100,那么就会有25个请求全部执行到if这里,可能就会重复的执行这个任务,所以我们要做一下限流
        // 保证每次只能有一个线程提交/执行这个任务
        int remainSize = localUnSeqIdBO.getIdQueue().size();
        // 使用的空间比总空间的25%还小,那么说明我用的id的空间超过了75%,这个时候就要考虑进行本地id段的更新了
        if (step * UPDATE_UN_SEQ_ID_RATE > remainSize) {
            Semaphore semaphore = semaphoreMap.get(localUnSeqIdBO.getId());
            if (semaphore == null) {
                LOGGER.error("semaphore is error  id is {}", localUnSeqIdBO.getId());
                return;
            }
            boolean acquireStatus = semaphore.tryAcquire();
            if (acquireStatus) {
                LOGGER.info("开始尝试进行本地id段的同步操作");
                threadPoolExecutor.execute(() -> {
                    // 尝试抢占id段 , 由于这块非常消耗网络IO-->需要update,select等操作->需要操作MySQL,十分耗费性能
                    // 如果同步执行(全部由主线程执行) , 那么可能就会导致主线程阻塞,所以我们可以让其进行异步执行,使用一个线程池
                    IdGeneratePO idGeneratePO = idGenerateMapper.selectById(localUnSeqIdBO.getId());
                    tryUpdateMySQLRecord(idGeneratePO);
                    semaphoreMap.get(idGeneratePO.getId()).release();
                    LOGGER.info("本地无序id段同步完成，id is {}", localUnSeqIdBO.getId());
                });
            }
        }
    }

    /**
     * 更新本地有序的id段
     *
     * @param localSeqIdBO
     */
    private void refreshLocalSeqId(LocalSeqIdBO localSeqIdBO) {
        // 得到当前id段总占用情况
        long step = localSeqIdBO.getNextThreshold() - localSeqIdBO.getCurrentStart();
        // 还有一种情况,假设并发量是100,那么就会有25个请求全部执行到if这里,可能就会重复的执行这个任务,所以我们要做一下限流
        // 保证每次只能有一个线程提交/执行这个任务
        if (localSeqIdBO.getCurrentNum().get() - localSeqIdBO.getCurrentStart()  > step * UPDATE_SEQ_ID_RATE) {
            Semaphore semaphore = semaphoreMap.get(localSeqIdBO.getId());
            if (semaphore == null) {
                LOGGER.error("semaphore is error  id is {}", localSeqIdBO.getId());
                return;
            }
            boolean acquireStatus = semaphore.tryAcquire();
            if (acquireStatus) {
                LOGGER.info("开始尝试进行本地id段的同步操作");
                threadPoolExecutor.execute(() -> {
                    // 尝试抢占id段 , 由于这块非常消耗网络IO-->需要update,select等操作->需要操作MySQL,十分耗费性能
                    // 如果同步执行(全部由主线程执行) , 那么可能就会导致主线程阻塞,所以我们可以让其进行异步执行,使用一个线程池
                    IdGeneratePO idGeneratePO = idGenerateMapper.selectById(localSeqIdBO.getId());
                    tryUpdateMySQLRecord(idGeneratePO);
                    semaphoreMap.get(idGeneratePO.getId()).release();
                    LOGGER.info("本地有序id段同步完成，id is {}", idGeneratePO.getId());
                });
            }
        }
    }

    /**
     * SpringBoot服务初始化的时候,就会触发这个afterPropertiesSet回调
     * 我们需要通过查询数据库配置表加载,保存到map中-->更新本地内存Map
     * 服务在每次生成分布式id的时候,会在本地内存缓存一个id段
     *
     * @throws Exception 异常
     */
    @Override
    public void afterPropertiesSet() throws Exception {
        LOGGER.info("服务刚刚启动,抢占新的id段");
        // 查询数据库id发号器配置表,保存到map对象中
        List<IdGeneratePO> idGeneratePOS = idGenerateMapper.selectAll();
        for (IdGeneratePO idGeneratePO : idGeneratePOS) {
            // 因为不同机器一定要使用不同的id段,所以我们机器要抢占id段,进行id段占用
            tryUpdateMySQLRecord(idGeneratePO);
            semaphoreMap.put(idGeneratePO.getId(), new Semaphore(1));
        }
    }

    /**
     * 尝试更新MySQL中的分布式id配置信息,占有相应的id段
     *
     * @param idGeneratePO PO对象
     */
    private void tryUpdateMySQLRecord(IdGeneratePO idGeneratePO) {
        int update = idGenerateMapper.updateNewIdCountAndVersion(idGeneratePO.getId(), idGeneratePO.getVersion());
        if (update > 0) {
            // 抢占id段成功,保存到本地map中
            createLocalIdBO(idGeneratePO);
            return;
        }
        // 如果抢占id段失败,那么重试三次
        for (int i = 0; i < 3; i++) {
            // 重新查询对应的分布式id策略
            idGeneratePO = idGenerateMapper.selectById(idGeneratePO.getId());
            update = idGenerateMapper.updateNewIdCountAndVersion(idGeneratePO.getId(), idGeneratePO.getVersion());
            if (update > 0) {
                createLocalIdBO(idGeneratePO);
                break;
            }
        }
        // 如果尝试三次还没有更新id段成功,那么就抛出异常
        throw new RuntimeException("表id段占用失败，竞争过于激烈，id is " + idGeneratePO.getId());
    }

    /**
     * 保存id到本地map中
     *
     * @param idGeneratePO
     */
    private static void createLocalIdBO(IdGeneratePO idGeneratePO) {
        long begin = idGeneratePO.getCurrentStart();
        long end = idGeneratePO.getNextThreshold();
        long currentNum = begin;
        if (idGeneratePO.getIsSeq() == ORDER_ID_TYPE) {
            // 抢占有序id段,保存到本地map中
            LocalSeqIdBO localSeqIdBO = new LocalSeqIdBO();
            AtomicLong atomicLong = new AtomicLong(currentNum);
            localSeqIdBO.setId(idGeneratePO.getId());
            localSeqIdBO.setCurrentNum(atomicLong);
            localSeqIdBO.setCurrentStart(begin);
            localSeqIdBO.setNextThreshold(end);
            // 保存到本地map中
            localSeqIdBOMap.put(localSeqIdBO.getId(), localSeqIdBO);
        } else {
            // 抢占无序id段,保存到本地map中
            LocalUnSeqIdBO localUnSeqIdBO = new LocalUnSeqIdBO();
            localUnSeqIdBO.setId(idGeneratePO.getId());
            ConcurrentLinkedQueue<Long> idQueue = new ConcurrentLinkedQueue<>();
            List<Long> ids = new ArrayList<>();
            for (long i = begin; i <= end; ++i) {
                ids.add(i);
            }
            // 打乱后,放进queue中
            Collections.shuffle(ids);
            idQueue.addAll(ids);
            localUnSeqIdBO.setIdQueue(idQueue);
            localUnSeqIdBO.setCurrentStart(begin);
            localUnSeqIdBO.setNextThreshold(end);
            localUnSeqIdBOMap.put(localUnSeqIdBO.getId(), localUnSeqIdBO);
        }
    }
}